package com.aboo.kafka.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.UUID;

@Slf4j
@Component
@EnableScheduling
public class KafkaProducer {

    @Qualifier
    private KafkaTemplate<String, String> kafkaTemplate;

    @Scheduled(cron = "0/2 * * * * ?")
    public void send() {
        String msg = UUID.randomUUID().toString();
        ListenableFuture listenableFuture = kafkaTemplate.send("kafka_test", msg);
        listenableFuture.addCallback(
                o -> log.info("消息发送成功,{}", o.toString()),
                throwable -> log.info("消息发送失败,{}" + throwable.getMessage())
        );
        log.info("msg:{}", msg);
    }

}
